package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}

object Demo2DSL {
  def main(args: Array[String]): Unit = {
    //1、创建Spark sql环境
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("sql")
      .config("spark.sql.shuffle.partitions", 1) //默认在集群中时200个
      .getOrCreate()


    /**
     * DataFrame: 底层就是RDD, 比RDD多了列名和列的类型
     */
    //2、读取数据
    val students: DataFrame = spark
      .read
      .schema("id STRING, name STRING, age INT, sex STRING , clazz STRING") //指定列名和类型
      .option("sep", ",") //指定列分隔符，csv格式默认时逗号
      .csv("spark/data/students.csv") //读取csv格式的数据


    //当一个DF多次使用时，可以进行缓存
    students.cache()

    /**
     * show: 打印结果，相当于一个action算子
     */
    students.show()
    students.show(10)
    students.show(false) //完整显示数据


    /**
     * select
     */

    //通过字段名选择字段
    students.select("id", "name").show()

    //导入隐式转换
    import spark.implicits._

    //导入spark sql的函数
    import org.apache.spark.sql.functions._

    //使用列对象的方式
    students.select($"id", $"age" + 100).show()
    //在选择字段时对字段进行处理
    students.select($"id", substring($"clazz", 1, 2) as "class").show()

    /**
     * where
     */

    //sql表达式
    students.where("age > 23 and sex = '女'").show()

    //列对象方式
    students.where($"age" === 22 and $"sex" =!= '女').show()


    /**
     * group by
     */

    students.groupBy("clazz").count().show()

    //使用列对象的方式
    students.groupBy(substring($"clazz", 1, 2) as "clazz").count().show()

    /**
     * agg
     * 聚合函数：sum max min avg count countDistinct
     *
     * DSl和SQL的区别，SQL中的聚合放在select中，DSl的聚合放在AGG中
     */
    students.agg(max($"age")).show()

    //分组聚合
    students
      .groupBy($"clazz")
      .agg(count($"id") as "num", max("age") as "maxAge")
      .show()


    /**
     * join
     */

    val scores: DataFrame = spark
      .read
      .schema("sid STRING, cid STRING,score DOUBLE") //指定列名和类型
      .option("sep", ",") //指定列分隔符，csv格式默认时逗号
      .csv("spark/data/score.txt") //读取csv格式的数据


    //当关联的字段名一样时
    //students.join(scores, "id").show()

    //当关联字段名不一样时
    //hint("broadcast") 广播小表
    students.join(scores.hint("broadcast"), $"id" === $"sid").show()

    students
      .join(scores, $"id" === $"sid")
      .groupBy("name")
      .agg(sum($"score") as "sumScore")
      .show()

    /**
     * explode
     */
    val linesDF: DataFrame = spark
      .read
      .schema("line STRING") //指定列名和类型
      .option("sep", "|") //指定列分隔符，csv格式默认时逗号
      .csv("spark/data/words.txt") //读取csv格式的数据

    linesDF
      //一行转换成多行
      .select(explode(split($"line", ",")) as "word")
      .groupBy($"word")
      .agg(count("word") as "count")
      .show()

    /**
     * 窗口函数
     * row_number rank lag lead max min count sum avg
     */


    import org.apache.spark.sql.expressions.Window._
    //在select中使用窗口函数
    students
      .select($"id", $"name", $"age", $"sex", $"clazz", row_number() over partitionBy($"clazz").orderBy($"age".desc) as "r")
      .show()


    //withColumn： 在原表的基础上增加列
    students
      .withColumn("r", dense_rank() over partitionBy($"clazz").orderBy($"age".desc))
      .show()

  }
}
